package producer.broker;

import api.Message;
import api.MessageType;
import org.checkerframework.checker.units.qual.A;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import producer.constant.BrokerMessageStatus;

/**
 * @Desc: ""
 * @Author: caixiang
 * @DATE: 2020/6/23 11:23
 */
public class RabbitBrokerImpl implements RabbitBroker {

    Logger logger = LoggerFactory.getLogger(RabbitBroker.class);


    @Autowired
    private RabbitTemplateContainer rabbitTemplateContainer;

    @Override
    public void rapidSend(Message message) {
        message.setMessageId(MessageType.RAPID);
        sendKernel(message);
    }

    /**
     *  @Description 发送消息的核心方法， 使用异步线程池进行消息发送
     *  @Param
     *  @Return  
     *  @Author caixiang
     *  @Date 2020/6/23  11:29 
     **/
    private void sendKernel(Message message) {
        AsyncBaseQueue.submit((Runnable)()->{
            CorrelationData correlationData = new CorrelationData(String.format("%s#%s",message.getMessageId(),System.currentTimeMillis()));
            String topic = message.getTopic();
            String routingKey = message.getRoutingKey();
            RabbitTemplate rabbitTemplate = rabbitTemplateContainer.getTemplate(message);
            rabbitTemplate.convertAndSend(topic,routingKey,message,correlationData);
            logger.info("# producer.broker.RabbitBrokerImpl # send to rabbitmq ,messageId:{}",message.getMessageId());
        });
    }


    @Override
    public void confirmSend(Message message) {
        message.setMessageType(MessageType.CONFIRM);
        sendKernel(message);
    }

    @Override
    public void reliantSend(Message message) {
        message.setMessageType(MessageType.RELIANT);
        //1.  先构造数据库Entity （并且赋值初始状态），，再把这个entity录到数据库
        /**
         * CREATE TABLE IF NOT EXISTS `broker_message`(
         * 	`message_id` varchar(128) NOT NULL,    //message.getMessageId()
         * 	`message` varchar(4000),
         * 	`try_count` int(4) DEFAULT 0,
         * 	`status` varchar(10) DEFAULT '',
         * 	`next_retry` TIMESTAMP NOT NULL DEFAULT '0000-00-00 00:00:00',  //能够容忍最大超时时间
         * 	`create_time` TIMESTAMP NOT NULL DEFAULT '0000-00-00 00:00:00',
         * 	`update_time` TIMESTAMP NOT NULL DEFAULT '0000-00-00 00:00:00',
         * 	PRIMARY KEY(`message_id`)
         * ) ENGINE=INNODB DEFAULT CHARSET=utf8;
         *
         * */
        //2.发送消息到MQ
        //发送消息给broker后，broker会给我一个ack，给我ack后我再去更新这条消息把BrokerMessageStatus.SENDING 改成SEND_OK 或者SEND_FAIL
        sendKernel(message);
    }

    @Override
    public void sendMessage() {

    }
}
